sqs source: json codec support to split sqs message into multiple events #5330
Conversation
…iple events Signed-off-by: Jeremy Michael <[email protected]>
jmsusanto
requested review from
sb2k16,
chenqi0805,
engechas,
graytaylor0,
dinujoh,
kkondaka,
KarstenSchnitter,
dlvenable and
oeyh
as code owners
jmsusanto
changed the title
| eventMetadata.setAttribute(lowerCamelCaseKey, entry.getValue()); | ||
| } | ||
|
|
||
| for (Map.Entry<String, MessageAttributeValue> entry : message.messageAttributes().entrySet()) { |
There was a problem hiding this comment.
A lot of this code is similar to the RawSqsMessageHandler. We should not duplicate this code.
I think the best option is to update the existing RawSqsMessageHandler to support an injectable message strategy.
It might look like:
interface MessageFieldStrategy {
List<Event> parseEvents(String messageBody);
}
This has some advantages of being able to use buffer.writeAll for the whole batch and the attribute code is shared for all.
There was a problem hiding this comment.
I removed the BulkMessageHandler and added strategies instead. buffer.writeAll is also used instead of buffer.write now
| for (Map.Entry<String, MessageAttributeValue> entry : message.messageAttributes().entrySet()) { | ||
| final String originalKey = entry.getKey(); | ||
| final String lowerCamelCaseKey = originalKey.substring(0, 1).toLowerCase() + originalKey.substring(1);; | ||
| eventMetadata.setAttribute(lowerCamelCaseKey, entry.getValue().stringValue()); |
There was a problem hiding this comment.
A lot of this processing can be shared, which has the added benefit of avoiding extra memory. You can do this in the refactoring that I suggest above.
That is, before the loop over the List<Event>, create a Map<String, String> for all the attributes. Then re-use those values. This should reduce both compute and memory.
There was a problem hiding this comment.
great idea, the metadata/attributes of a bulk message would remain the same for every event in that message
| @@ -0,0 +1,73 @@ | |||
| /* | |||
There was a problem hiding this comment.
Please use the longer copyright header.
/*
* Copyright OpenSearch Contributors
* SPDX-License-Identifier: Apache-2.0
*
* The OpenSearch Contributors require contributions made to
* this file be licensed under the Apache-2.0 license or a
* compatible open source license.
*
*/
There was a problem hiding this comment.
added to every file
Signed-off-by: Jeremy Michael <[email protected]>
Signed-off-by: Jeremy Michael <[email protected]>
There was a problem hiding this comment.
Thank you @jmsusanto . This is looking good. I have a few smaller comments.
| import java.util.List; | ||
| import java.util.function.Consumer; | ||
|
|
||
| public class JsonBulkMessageFieldStrategy implements MessageFieldStrategy { |
There was a problem hiding this comment.
Just rename this to CodecBulkMessageFieldStrategy because it is not bound to JSON.
| } else { | ||
| sqsEventProcessor = new SqsEventProcessor(new RawSqsMessageHandler()); | ||
| MessageFieldStrategy standardStrategy = new StandardMessageFieldStrategy(); | ||
| sqsEventProcessor = new SqsEventProcessor(new RawSqsMessageHandler(standardStrategy)); |
There was a problem hiding this comment.
This line and line 89 are the same. Refactor to use the same line. Just get the MessageFieldStrategy from the conditional.
There was a problem hiding this comment.
that makes sense, i made the change
| import java.util.List; | ||
|
|
||
| public interface MessageFieldStrategy { | ||
| // Convert the SQS message body into one or more events. |
There was a problem hiding this comment.
Make this into a Javadoc comment.
Signed-off-by: Jeremy Michael <[email protected]>
Description
adds json codec support and functionality to split message into multiple events
#5054
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.